#include "dtu_mqtt.h"
#include "dtu_log.h"
#include "dtu_memory.h"
#include "dtu_config.h"
#include "dtu_data_collect.h"
#include "dtu_at_command.h"
#include "MQTTClient.h"
#include <string.h>
#include <time.h>
#include <stdio.h>
#include <stdlib.h>

/* MQTT客户端结构 */
typedef struct {
    MQTTClient client;
    char client_id[256];  /* 增大缓冲区以支持长客户端ID */
    char broker_url[256];
    bool connected;
    bool running;
    pthread_mutex_t mutex;
    mqtt_config_t config;
    mqtt_stats_t stats;
    void (*message_callback)(const mqtt_message_t* message, void* user_data);
    void* callback_user_data;
    time_t last_connect_time;
    time_t last_ping_time;
} mqtt_client_paho_t;

/* 全局MQTT客户端实例 */
static mqtt_client_paho_t g_mqtt_client = {0};

/* 内部函数声明 */
static void connection_lost_callback(void* context, char* cause);
static int message_arrived_callback(void* context, char* topicName, int topicLen, MQTTClient_message* message);
static void delivery_complete_callback(void* context, MQTTClient_deliveryToken token);
static void handle_mqtt_command_message(const char* topic, const char* payload, int payload_len);
static void handle_simple_poll_command(const char* poll_string);
static void send_simple_poll_response(const char* poll_string, dtu_error_t result, const char* response_data);

/* 初始化MQTT客户端 */
dtu_error_t dtu_mqtt_init(const mqtt_config_t* config)
{
    if (!config) {
        return DTU_ERROR_INVALID_PARAM;
    }

    pthread_mutex_lock(&g_mqtt_client.mutex);

    /* 复制配置 */
    memcpy(&g_mqtt_client.config, config, sizeof(mqtt_config_t));

    /* 使用配置的完整客户端ID，不做任何修改 */
    DTU_STRNCPY(g_mqtt_client.client_id, config->client_id, sizeof(g_mqtt_client.client_id));

    /* 构建broker URL */
    snprintf(g_mqtt_client.broker_url, sizeof(g_mqtt_client.broker_url),
             "tcp://%s:%d", config->broker, config->port);

    /* 创建MQTT客户端 */
    int rc = MQTTClient_create(&g_mqtt_client.client, g_mqtt_client.broker_url,
                               g_mqtt_client.client_id, MQTTCLIENT_PERSISTENCE_NONE, NULL);
    
    if (rc != MQTTCLIENT_SUCCESS) {
        pthread_mutex_unlock(&g_mqtt_client.mutex);
        DTU_LOG_ERROR("Failed to create MQTT client: %d", rc);
        return DTU_ERROR;
    }

    /* 设置回调函数 */
    MQTTClient_setCallbacks(g_mqtt_client.client, NULL, 
                           connection_lost_callback, 
                           message_arrived_callback, 
                           delivery_complete_callback);

    g_mqtt_client.connected = false;
    g_mqtt_client.running = false;
    
    pthread_mutex_unlock(&g_mqtt_client.mutex);

    DTU_LOG_INFO("MQTT client initialized with broker: %s, client_id: %s", 
                 g_mqtt_client.broker_url, g_mqtt_client.client_id);
    
    return DTU_OK;
}

/* 连接到MQTT服务器 */
dtu_error_t dtu_mqtt_connect(void)
{
    pthread_mutex_lock(&g_mqtt_client.mutex);

    if (g_mqtt_client.connected) {
        pthread_mutex_unlock(&g_mqtt_client.mutex);
        return DTU_OK;
    }

    /* 设置连接选项 */
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    conn_opts.keepAliveInterval = g_mqtt_client.config.keepalive;
    conn_opts.cleansession = g_mqtt_client.config.clean_session;  /* 使用配置的clean_session值 */
    conn_opts.reliable = 1;
    conn_opts.MQTTVersion = MQTTVERSION_3_1_1;  /* 使用MQTT 3.1.1版本 */

    /* 设置用户名和密码 */
    if (g_mqtt_client.config.username[0] != '\0') {
        conn_opts.username = g_mqtt_client.config.username;
    }
    if (g_mqtt_client.config.password[0] != '\0') {
        conn_opts.password = g_mqtt_client.config.password;
    }

    /* 设置连接超时 */
    conn_opts.connectTimeout = 30;

    DTU_LOG_INFO("Connecting to MQTT broker: %s with client_id: %s",
                 g_mqtt_client.broker_url, g_mqtt_client.client_id);
    DTU_LOG_INFO("MQTT connection parameters - keepalive: %d, clean_session: %d, username: %s",
                 conn_opts.keepAliveInterval, conn_opts.cleansession,
                 conn_opts.username ? conn_opts.username : "none");

    /* 连接到服务器 */
    int rc = MQTTClient_connect(g_mqtt_client.client, &conn_opts);

    if (rc != MQTTCLIENT_SUCCESS) {
        pthread_mutex_unlock(&g_mqtt_client.mutex);
        const char* error_msg = "Unknown error";
        switch (rc) {
            case 1: error_msg = "Connection refused: unacceptable protocol version"; break;
            case 2: error_msg = "Connection refused: identifier rejected"; break;
            case 3: error_msg = "Connection refused: server unavailable"; break;
            case 4: error_msg = "Connection refused: bad user name or password"; break;
            case 5: error_msg = "Connection refused: not authorized"; break;
            case 135: error_msg = "MQTT 5.0 not supported or protocol version mismatch"; break;
            case -1: error_msg = "Network connection failed"; break;
            case -2: error_msg = "Invalid parameters"; break;
            default: error_msg = "Connection failed"; break;
        }
        DTU_LOG_ERROR("Failed to connect to MQTT broker: %s (code: %d)", error_msg, rc);
        return DTU_ERROR;
    }

    g_mqtt_client.connected = true;
    g_mqtt_client.running = true;
    g_mqtt_client.last_connect_time = time(NULL);
    g_mqtt_client.last_ping_time = time(NULL);
    g_mqtt_client.stats.connect_count++;

    pthread_mutex_unlock(&g_mqtt_client.mutex);

    DTU_LOG_INFO("MQTT client connected successfully");

    /* 连接成功后自动订阅配置的Topic */
    const dtu_config_t* config = dtu_config_get();
    if (config && config->mqtt_subscribe_config.topic[0] != '\0') {
        dtu_error_t sub_ret = dtu_mqtt_subscribe(config->mqtt_subscribe_config.topic, 
                                                config->mqtt_subscribe_config.qos,
                                                NULL, NULL);
        if (sub_ret == DTU_OK) {
            DTU_LOG_INFO("Successfully subscribed to topic: %s", config->mqtt_subscribe_config.topic);
        } else {
            DTU_LOG_ERROR("Failed to subscribe to topic: %s", config->mqtt_subscribe_config.topic);
        }
    }

    return DTU_OK;
}

/* 断开MQTT连接 */
void dtu_mqtt_disconnect(void)
{
    pthread_mutex_lock(&g_mqtt_client.mutex);

    if (!g_mqtt_client.connected) {
        pthread_mutex_unlock(&g_mqtt_client.mutex);

    }

    /* 断开连接 */
    int rc = MQTTClient_disconnect(g_mqtt_client.client, 10000);
    
    if (rc != MQTTCLIENT_SUCCESS) {
        DTU_LOG_ERROR("Failed to disconnect MQTT client: %d", rc);
    }

    g_mqtt_client.connected = false;
    g_mqtt_client.running = false;

    pthread_mutex_unlock(&g_mqtt_client.mutex);

    DTU_LOG_INFO("MQTT client disconnected");
}

/* 发布消息 */
dtu_error_t dtu_mqtt_publish(const char* topic, const void* payload, size_t length,
                             mqtt_qos_t qos, bool retained)
{
    if (!topic || !payload) {
        return DTU_ERROR_INVALID_PARAM;
    }

    pthread_mutex_lock(&g_mqtt_client.mutex);

    if (!g_mqtt_client.connected) {
        pthread_mutex_unlock(&g_mqtt_client.mutex);
        DTU_LOG_ERROR("MQTT client not connected");
        return DTU_ERROR;
    }

    /* 创建消息 */
    MQTTClient_message pubmsg = MQTTClient_message_initializer;
    pubmsg.payload = (void*)payload;
    pubmsg.payloadlen = length;
    pubmsg.qos = (int)qos;
    pubmsg.retained = retained;

    MQTTClient_deliveryToken token;

    /* 发布消息 */
    int rc = MQTTClient_publishMessage(g_mqtt_client.client, topic, &pubmsg, &token);
    
    if (rc != MQTTCLIENT_SUCCESS) {
        pthread_mutex_unlock(&g_mqtt_client.mutex);
        DTU_LOG_ERROR("Failed to publish MQTT message to topic: %s, error: %d", topic, rc);
        g_mqtt_client.stats.disconnect_count++;
        return DTU_ERROR;
    }

    /* 等待消息发送完成 */
    rc = MQTTClient_waitForCompletion(g_mqtt_client.client, token, 5000);

    if (rc != MQTTCLIENT_SUCCESS) {
        pthread_mutex_unlock(&g_mqtt_client.mutex);
        DTU_LOG_ERROR("Failed to wait for MQTT message completion: %d", rc);
        g_mqtt_client.stats.disconnect_count++;
        return DTU_ERROR;
    }

    g_mqtt_client.stats.messages_sent++;
    g_mqtt_client.stats.bytes_sent += length;
    g_mqtt_client.stats.last_message_time = time(NULL);

    pthread_mutex_unlock(&g_mqtt_client.mutex);

    /* 简化的MQTT发布日志，只在DEBUG级别显示 */
    DTU_LOG_DEBUG("MQTT published: %zu bytes", length);
    return DTU_OK;
}

/* 订阅主题 */
dtu_error_t dtu_mqtt_subscribe(const char* topic, mqtt_qos_t qos,
                               void (*callback)(const mqtt_message_t* message, void* user_data),
                               void* user_data)
{
    if (!topic) {
        return DTU_ERROR_INVALID_PARAM;
    }

    pthread_mutex_lock(&g_mqtt_client.mutex);

    if (!g_mqtt_client.connected) {
        pthread_mutex_unlock(&g_mqtt_client.mutex);
        DTU_LOG_ERROR("MQTT client not connected");
        return DTU_ERROR;
    }

    /* 设置消息回调 */
    if (callback) {
        g_mqtt_client.message_callback = callback;
        g_mqtt_client.callback_user_data = user_data;
    }

    /* 订阅主题 */
    int rc = MQTTClient_subscribe(g_mqtt_client.client, topic, (int)qos);
    
    if (rc != MQTTCLIENT_SUCCESS) {
        pthread_mutex_unlock(&g_mqtt_client.mutex);
        DTU_LOG_ERROR("Failed to subscribe to MQTT topic: %s, error: %d", topic, rc);
        return DTU_ERROR;
    }

    g_mqtt_client.stats.messages_sent++;

    pthread_mutex_unlock(&g_mqtt_client.mutex);

    DTU_LOG_INFO("Successfully subscribed to MQTT topic: %s", topic);
    return DTU_OK;
}

/* 取消订阅主题 */
dtu_error_t dtu_mqtt_unsubscribe(const char* topic)
{
    if (!topic) {
        return DTU_ERROR_INVALID_PARAM;
    }

    pthread_mutex_lock(&g_mqtt_client.mutex);

    if (!g_mqtt_client.connected) {
        pthread_mutex_unlock(&g_mqtt_client.mutex);
        return DTU_ERROR;
    }

    /* 取消订阅 */
    int rc = MQTTClient_unsubscribe(g_mqtt_client.client, topic);
    
    if (rc != MQTTCLIENT_SUCCESS) {
        pthread_mutex_unlock(&g_mqtt_client.mutex);
        DTU_LOG_ERROR("Failed to unsubscribe from MQTT topic: %s, error: %d", topic, rc);
        return DTU_ERROR;
    }

    pthread_mutex_unlock(&g_mqtt_client.mutex);

    DTU_LOG_INFO("Successfully unsubscribed from MQTT topic: %s", topic);
    return DTU_OK;
}

/* 检查连接状态 */
bool dtu_mqtt_is_connected(void)
{
    pthread_mutex_lock(&g_mqtt_client.mutex);
    bool connected = g_mqtt_client.connected && MQTTClient_isConnected(g_mqtt_client.client);
    pthread_mutex_unlock(&g_mqtt_client.mutex);
    return connected;
}

/* 获取统计信息 */
dtu_error_t dtu_mqtt_get_stats(mqtt_stats_t* stats)
{
    if (!stats) {
        return DTU_ERROR_INVALID_PARAM;
    }

    pthread_mutex_lock(&g_mqtt_client.mutex);
    memcpy(stats, &g_mqtt_client.stats, sizeof(mqtt_stats_t));
    pthread_mutex_unlock(&g_mqtt_client.mutex);

    return DTU_OK;
}

/* 清理MQTT客户端 */
void dtu_mqtt_cleanup(void)
{
    pthread_mutex_lock(&g_mqtt_client.mutex);

    if (g_mqtt_client.connected) {
        MQTTClient_disconnect(g_mqtt_client.client, 1000);
        g_mqtt_client.connected = false;
    }

    if (g_mqtt_client.client) {
        MQTTClient_destroy(&g_mqtt_client.client);
        g_mqtt_client.client = NULL;
    }

    g_mqtt_client.running = false;

    pthread_mutex_unlock(&g_mqtt_client.mutex);

    DTU_LOG_INFO("MQTT client cleaned up");
}

/* ========== 回调函数实现 ========== */

/* 连接丢失回调 */
static void connection_lost_callback(void* context, char* cause)
{
    (void)context;

    pthread_mutex_lock(&g_mqtt_client.mutex);
    g_mqtt_client.connected = false;
    g_mqtt_client.stats.disconnect_count++;
    pthread_mutex_unlock(&g_mqtt_client.mutex);

    DTU_LOG_WARN("MQTT connection lost: %s", cause ? cause : "Unknown reason");

    /* TODO: 实现自动重连逻辑 */
}

/* 消息到达回调 */
static int message_arrived_callback(void* context, char* topicName, int topicLen, MQTTClient_message* message)
{
    (void)context;
    (void)topicLen;

    if (!topicName || !message) {
        return 1;
    }

    pthread_mutex_lock(&g_mqtt_client.mutex);
    g_mqtt_client.stats.messages_received++;
    g_mqtt_client.stats.bytes_received += message->payloadlen;
    g_mqtt_client.stats.last_message_time = time(NULL);
    pthread_mutex_unlock(&g_mqtt_client.mutex);

    DTU_LOG_INFO("MQTT message received - Topic: %s, Payload length: %d",
                 topicName, message->payloadlen);

    /* 打印消息内容（限制长度） */
    if (message->payloadlen > 0 && message->payload) {
        int print_len = message->payloadlen > 200 ? 200 : message->payloadlen;
        DTU_LOG_INFO("MQTT message payload: %.*s%s",
                     print_len, (char*)message->payload,
                     message->payloadlen > 200 ? "..." : "");
    }

    /* 调用用户回调函数 */
    if (g_mqtt_client.message_callback) {
        mqtt_message_t msg = {0};
        DTU_STRNCPY(msg.topic, topicName, sizeof(msg.topic));
        msg.payload_length = message->payloadlen;
        if (message->payloadlen > 0 && (size_t)message->payloadlen < sizeof(msg.payload)) {
            memcpy(msg.payload, message->payload, message->payloadlen);
            msg.payload[message->payloadlen] = '\0';
        }
        msg.qos = message->qos;
        msg.retained = message->retained;

        g_mqtt_client.message_callback(&msg, g_mqtt_client.callback_user_data);
    }

    /* 处理接收到的命令消息 */
    handle_mqtt_command_message(topicName, (char*)message->payload, message->payloadlen);

    /* 释放消息 */
    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);

    return 1;
}

/* 消息发送完成回调 */
static void delivery_complete_callback(void* context, MQTTClient_deliveryToken token)
{
    (void)context;
    DTU_LOG_DEBUG("MQTT message delivery completed, token: %d", token);
}

/* 处理MQTT命令消息 */
static void handle_mqtt_command_message(const char* topic, const char* payload, int payload_len)
{
    if (!topic || !payload || payload_len <= 0) {
        return;
    }

    DTU_LOG_INFO("Processing MQTT poll command from topic: %s", topic);

    /* 创建以null结尾的字符串 */
    char* poll_string = DTU_MALLOC(payload_len + 1);
    if (!poll_string) {
        DTU_LOG_ERROR("Failed to allocate memory for poll string");
        return;
    }
    memcpy(poll_string, payload, payload_len);
    poll_string[payload_len] = '\0';

    /* 去除可能的换行符和空格 */
    char* trimmed = poll_string;
    while (*trimmed && (*trimmed == ' ' || *trimmed == '\t' || *trimmed == '\n' || *trimmed == '\r')) {
        trimmed++;
    }

    int len = strlen(trimmed);
    while (len > 0 && (trimmed[len-1] == ' ' || trimmed[len-1] == '\t' ||
                       trimmed[len-1] == '\n' || trimmed[len-1] == '\r')) {
        trimmed[--len] = '\0';
    }

    if (len == 0) {
        DTU_LOG_ERROR("Empty poll string received");
        DTU_FREE(poll_string);
        return;
    }

    DTU_LOG_INFO("Received poll string: %s", trimmed);

    /* 直接处理16进制轮询字符串 */
    handle_simple_poll_command(trimmed);

    DTU_FREE(poll_string);
}

/* 处理简化的轮询采集命令 */
static void handle_simple_poll_command(const char* poll_string)
{
    if (!poll_string || strlen(poll_string) == 0) {
        DTU_LOG_ERROR("Invalid poll string");
        return;
    }

    DTU_LOG_INFO("Processing simple poll command: %s", poll_string);

    /* 验证16进制字符串格式 */
    int len = strlen(poll_string);
    if (len % 2 != 0) {
        DTU_LOG_ERROR("Invalid hex string length: %d", len);
        send_simple_poll_response(poll_string, DTU_ERROR, "Invalid hex string format");
        return;
    }

    for (int i = 0; i < len; i++) {
        char c = poll_string[i];
        if (!((c >= '0' && c <= '9') || (c >= 'A' && c <= 'F') || (c >= 'a' && c <= 'f'))) {
            DTU_LOG_ERROR("Invalid hex character: %c", c);
            send_simple_poll_response(poll_string, DTU_ERROR, "Invalid hex character");
            return;
        }
    }

    /* 构造AT+POLLSTR命令
     * 使用默认参数：串口ID=1(RS485), 设备ID=1, 字符串编号=1, CRC使能=1
     */
    char at_command[512];
    snprintf(at_command, sizeof(at_command),
             "AT+POLLSTR=1,1,1,1,\"%s\"", poll_string);

    DTU_LOG_INFO("Executing AT command: %s", at_command);

    /* 执行AT命令 */
    at_cmd_response_t response = {0};
    dtu_error_t result = dtu_at_execute(at_command, &response);

    if (result == DTU_OK) {
        DTU_LOG_INFO("Poll command executed successfully");
        send_simple_poll_response(poll_string, result, response.response);
    } else {
        DTU_LOG_ERROR("Poll command execution failed");
        send_simple_poll_response(poll_string, result, "AT command execution failed");
    }
}

/* 发送简化的轮询响应到MQTT */
static void send_simple_poll_response(const char* poll_string, dtu_error_t result, const char* response_data)
{
    const dtu_config_t* config = dtu_config_get();
    if (!config || strlen(config->push_config.topic) == 0) {
        DTU_LOG_ERROR("MQTT publish topic not configured");
        return;
    }

    /* 构造简单的响应JSON字符串 */
    char json_response[2048];
    snprintf(json_response, sizeof(json_response),
             "{"
             "\"timestamp\":%ld,"
             "\"poll_string\":\"%s\","
             "\"success\":%s",
             time(NULL),
             poll_string ? poll_string : "",
             result == DTU_OK ? "true" : "false");

    /* 添加响应数据 */
    if (response_data && strlen(response_data) > 0) {
        char temp[1024];
        snprintf(temp, sizeof(temp), ",\"data\":\"%s\"", response_data);
        strncat(json_response, temp, sizeof(json_response) - strlen(json_response) - 1);
    }

    strncat(json_response, "}", sizeof(json_response) - strlen(json_response) - 1);

    /* 发送到MQTT */
    dtu_error_t publish_result = dtu_mqtt_publish(config->push_config.topic,
                                                 json_response, strlen(json_response),
                                                 1, false);
    if (publish_result == DTU_OK) {
        DTU_LOG_INFO("Simple poll response sent to MQTT - Success: %s",
                    result == DTU_OK ? "true" : "false");
        DTU_LOG_DEBUG("Response data: %s", json_response);
    } else {
        DTU_LOG_ERROR("Failed to send simple poll response to MQTT");
    }
}








